package com.weixin.project.Service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.weixin.project.Common.Enum.RatingScore;
import com.weixin.project.Config.Config;
import com.weixin.project.Entity.Question;
import com.weixin.project.Entity.QuestionEmb;
import com.weixin.project.Entity.UserEmb;
import com.weixin.project.Util.CollectionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@SuppressWarnings({"unchecked", "rawtypes"})
@Service
public class RecommendService {
    @Value("${python.recommendURL}")
    private String pythonURL;
    @Autowired
    QuestionService questionService;
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    EmbService embService;
    @Autowired
    PracticeService practiceService;
    @Autowired
    RestTemplate restTemplate;
    @Autowired
    HttpService httpService;
    //每个embedding的大小
    private final static Integer embeddingSize = 32;
    //相似度阈值
    private final static Double threshold = 0.6;
    //前端每道题推荐的相似题目数量
    private final static Integer similarQuestionCount = 3;
    //根据用户最近做的K个题目进行在线推荐
    private final static Integer K = 20;
    //实时召回题目集合的size
    private final static Integer onlineRecallSize = 40;
    //离线召回题目集合的size
    private final static Integer offlineRecallSize = 60;
    //召回题目集合的size
    private final static Integer recallSize = onlineRecallSize + offlineRecallSize;
    //随机推题推荐的题目数量
    private final static Integer recommendSize = 10;


    //hash，用户对题目的”打分“，无expire，定时任务清理，需要借助TIME_OF_BEHAVIOR_ZSET维护
    private final static String RATING_OF_USER_HASH = "rating_of_user";
    //zset，key为用户做题的题目id，score为timestamp，无expire，定时任务清理
    private final static String TIME_OF_BEHAVIOR_ZSET = "time_of_behavior";
    //zset，用户的线上推荐题目列表，key为题目id，score为推荐优先级，无expire，定时任务清理
    private final static String ONLINE_CANDIDATES_OF_USER = "online_candidates_of_user";
    
    /**
     * 用户练习过的题目
     * set, setKey = USER_PRACTICED_QUESTIONS + "::" + userId
     * element = questionId
     */
    private final static String USER_PRACTICED_QUESTIONS = PracticeService.USER_PRACTICED_QUESTIONS;

    /**
     * 相似题目
     * hash, hashKey = SIMILAR_QUESTIONS_OF_QUESTION + "::" + questionId
     * key = "key::" + questionId
     * value = cosSimilarity
     * expire: 30min
     */
    private final static String SIMILAR_QUESTIONS_OF_QUESTION = "similar_questions_of_question";

    /**
     * 离线召回集合
     * hash, hashKey = OFFLINE_CANDIDATES_OF_USER + "::" + userId
     * key = "key::" + questionId
     * value = cosSimilarity
     * expire: 30min
     */
    private final static String OFFLINE_CANDIDATES_OF_USER = "offline_candidates_of_user";

    public List<Question> recommend(Integer userId) {
        if (!Config.REC_IS_ON){
            //随机抽10个题目
            Set<Integer> set = CollectionUtil.randomSubSet(questionService.allQuestionsSet(), 10);
            return set.stream().map(x -> questionService.findQuestionById(x)).collect(Collectors.toList());
        }
        return rank(userId, retrieval(userId));
    }

    /**
     * 排序层，使用LR
     * @param userId 用户自增Id
     * @return 排序结果，推荐列表，固定为10个题目
     */
    private List<Question> rank(Integer userId, Set<Integer> questions){
        //这个分支不应该被进入，但有助于保证代码的鲁棒性
        if (questions.size() < recommendSize){
            questions.addAll(CollectionUtil.randomSubSet(
                    CollectionUtil.diff(questionService.allQuestionsSet(), questions),
                    recommendSize - questions.size()
            ));
        }
        List<Question> ret = new ArrayList<>();
        String baseURL = pythonURL + "/rank";
        Map<String, Object> params = new HashMap<>();
        params.put("user_id", userId);
        params.put("candidates", questions);
        JSONObject object = httpService.sendGETRequest(baseURL, params);
        JSONArray results = object.getJSONArray("results");
        for (int i = 0; i < results.size(); i++) {
            ret.add(questionService.findQuestionById(results.getInteger(i)));
        }
        return ret;
    }


    /**
     * 增量训练LR模型，通过调用flask推荐服务接口实现
     */
    public void incLR(){
        Map<Integer, Set<Integer>> user_questions = practiceService.findPracticedQuestionsInOneHour();
        for (Integer userId : user_questions.keySet()){
            incLR(userId, user_questions.get(userId));
        }
    }

    private void incLR(Integer userId, Set<Integer> questions){
        JSONObject requestBody = new JSONObject();
        requestBody.put("user_id", userId);
        requestBody.put("positive",questions);
        requestBody.put("negative", CollectionUtil.randomSubSet(
                CollectionUtil.diff(questionService.allQuestionsSet(), practiceService.findUserPracticedQuestionSet(userId)),
                questions.size()));
        String url = pythonURL + "/inc_LR";
        httpService.sendPostRequest(url, requestBody);
    }

    /**
     * 召回层，目前实现方法只有多路召回
     * @param userId
     * @return
     */
    private Set<Integer> retrieval(Integer userId){
        return multiRetrieval(userId);
    }

    /**
     * 多路召回
     * @param userId 用户自增Id
     * @return 召回结果，questionId Set
     */
    private Set<Integer> multiRetrieval(Integer userId){
        if (Config.REC == Config.OFFLINE_ONLINE) {
            System.out.println("offlineRetrieval:" + offlineRetrieval(userId));
            System.out.println("onlineRetrieval:" + onlineRetrieval(userId));
            return CollectionUtil.diff(CollectionUtil.union(offlineRetrieval(userId), onlineRetrieval(userId)), excludeSet(userId));
        }
        else if (Config.REC == Config.OFFLINE)
            return CollectionUtil.diff(offlineRetrieval(userId), excludeSet(userId));
        else if (Config.REC == Config.ONLINE)
            return CollectionUtil.diff(onlineRetrieval(userId), excludeSet(userId));
        else {
            //理论上不可能到达这一分支
            System.out.println("请检查Config.REC!");
            return new HashSet<>();
        }
    }

    /**
     * 在线召回，返回在redis中维护的OnlineCandidatesZSET
     * @param userId user自增Id
     * @return 实时推荐集合
     */
    private Set<Integer> onlineRetrieval(Integer userId){
        String zsetKey = ONLINE_CANDIDATES_OF_USER + "::" + userId;
        if(!redisTemplate.hasKey(zsetKey)) return new HashSet<>();
        Set<String> range = redisTemplate.boundZSetOps(zsetKey).range(0, -1);
        Set<Integer> set = range.stream().map(x -> Integer.parseInt(x.split("::")[1])).collect(Collectors.toSet());
        return set;
    }

    /**
     * 离线召回，使用每日python端储存在redis中的结果
     * @param userId 用户id
     * @return questionId集合
     */
    private Set<Integer> offlineRetrieval(Integer userId){
        loadOfflineCandidates2Redis(userId);
        String hashKey = OFFLINE_CANDIDATES_OF_USER + "::" + userId;
        Map<Integer, Double> map = hash2Map(hashKey);
        return map.keySet();
    }

    @SuppressWarnings("Duplicates")
    private void loadOfflineCandidates2Redis(Integer userId){
        String hashKey = OFFLINE_CANDIDATES_OF_USER + "::" + userId;
        if (redisTemplate.hasKey(hashKey)){
            redisTemplate.expire(hashKey, 30, TimeUnit.MINUTES);
            return;
        }
        else {
            Map<String, Object> params = new HashMap<>();
            UserEmb userEmb = embService.findUserEmb(userId);
            if (userEmb != null) {
                return;
            }
            params.put("emb", userEmb.getEmbedding());
            String baseURL = pythonURL + "/offline_recall";
            JSONObject object = httpService.sendGETRequest(baseURL, params);
            if (!object.getBoolean("success")) return;
            JSONArray results = object.getJSONArray("results");
            System.out.println("results:" + results);
            for (int i = 0; i < results.size(); i++) {
                String key = "key::" + results.getJSONArray(i).getInteger(0);
                redisTemplate.boundHashOps(hashKey).put(key, results.getJSONArray(i).getDouble(1));
            }
            redisTemplate.expire(hashKey, 30, TimeUnit.MINUTES);
        }
    }

    /**
     * 获取用户已经练习过的题目集合，在多路召回时需要排除在外
     * @param userId 用户的自增Id
     * @return 用户已经做过的题目集合
     */
    private Set<Integer> excludeSet(Integer userId){
        String key = USER_PRACTICED_QUESTIONS + "::" + userId;
        if(!redisTemplate.hasKey(key)) {
            practiceService.loadUserPractices2Redis(userId);
        }
        System.out.println("excludeSet:" + redisTemplate.boundSetOps(key).members());
        return redisTemplate.boundSetOps(key).members();
    }

    private List<Double> findUserEmb(Integer id){
        UserEmb userEmb = embService.findUserEmb(id);
        if (userEmb == null){
            return null;
        }
        else{
            String[] split = userEmb.getEmbedding().split(",");
            List<String> list = Arrays.asList(split);
            return list.stream().map(x -> Double.parseDouble(x)).collect(Collectors.toList());
        }
    }

    private List<Double> findQuestionEmb(Integer id){
        QuestionEmb questionEmb = embService.findQuestionEmb(id);
        if (questionEmb == null){
            return null;
        }
        else{
            String[] split = questionEmb.getEmbedding().split(",");
            List<String> list = Arrays.asList(split);
            return list.stream().map(x -> Double.parseDouble(x)).collect(Collectors.toList());
        }
    }

    /**
     * 寻找相似题目，直接供前端展示用
     * @param questionId 问题Id
     * @return 相似题目列表
     */
    public List<Question> findSimilarQuestionsList(Integer questionId){
        if (!Config.REC_IS_ON){
            List<Question> questions = new ArrayList<>();
//            questions.add(questionService.findQuestionById(32528));
//            questions.add(questionService.findQuestionById(32529));
//            questions.add(questionService.findQuestionById(32530));
            return questions;
        }
        return findSimilarQuestionsList(questionId, similarQuestionCount);
    }

    private List<Question> findSimilarQuestionsList(Integer questionId, Integer num){
        Map<Integer, Double> map = findSimilarQuestionsMap(questionId);
        List<Map.Entry<Integer, Double>> list = map.entrySet().stream().
                sorted((p2, p1) -> p1.getValue().compareTo(p2.getValue())).collect(Collectors.toList());
        List<Question> questions = list.subList(0, num).stream().
                map(x -> questionService.findQuestionById(x.getKey())).collect(Collectors.toList());
        return questions;
    }

    /**
     * 寻找相似题目，以map的形式返回供后续处理，默认为20个
     * @param questionId 问题Id
     * @return 一个map，形式为{questionId: cosSimilarity}
     */
    private Map<Integer,Double> findSimilarQuestionsMap(Integer questionId){
        loadSimilarQuestions2Redis(questionId);
        String hashKey = SIMILAR_QUESTIONS_OF_QUESTION + "::" + questionId;
        return hash2Map(hashKey);
    }

    @SuppressWarnings("Duplicates")
    private void loadSimilarQuestions2Redis(Integer questionId){
        String hashKey = SIMILAR_QUESTIONS_OF_QUESTION + "::" + questionId;
        if (redisTemplate.hasKey(hashKey)) {
            redisTemplate.expire(hashKey, 30, TimeUnit.MINUTES);
            return;
        }
        else {
            Map<String, Object> params = new HashMap<>();
            params.put("question_id", questionId);
            String baseURL = pythonURL + "/similar_questions";
            JSONObject object = httpService.sendGETRequest(baseURL, params);
            JSONArray results = object.getJSONArray("results");
            System.out.println("results:" + results);
            for (int i = 0; i < results.size(); i++) {
                String key = "key::" + results.getJSONArray(i).getInteger(0);
                redisTemplate.boundHashOps(hashKey).put(key, results.getJSONArray(i).getDouble(1));
            }
            redisTemplate.expire(hashKey, 30, TimeUnit.MINUTES);
        }
    }

    @Async
    public void collectUserBehaviorForOnlineRec(Integer userId, Integer questionId, RatingScore ratingScore){
        updateOnlineRecSet(userId, questionId);
        maintainUserRating(userId, questionId, ratingScore);
        System.out.println("The recommend system has collected the behaviors of user.");
    }

    /**
     * 维护用户onlineRec题目集合
     * @param userId user的自增Id
     * @param questionId 用户最新一次交互的questionId
     */
    private void updateOnlineRecSet(Integer userId, Integer questionId){
        String zsetKey = ONLINE_CANDIDATES_OF_USER + "::" + userId;
        String hashKey = RATING_OF_USER_HASH + "::" + userId;
        Map<Integer,Double> candidates = findSimilarQuestionsMap(questionId);
        //用户最近做过的K个题目集合（实际size小于K也没有关系）
        Map<Integer, Double> userRecentK = hash2Map(hashKey);
        for (Integer qid: candidates.keySet()){
            Double priority = calculatePriority(qid, userRecentK);
            System.out.println("priority:" + priority);
            String key = "key::" + qid;
            redisTemplate.boundZSetOps(zsetKey).add(key, priority);
            if (redisTemplate.boundZSetOps(zsetKey).size() > onlineRecallSize){
                redisTemplate.boundZSetOps(zsetKey).popMin();
            }
        }
    }

    /**
     * 维护在线推荐所需要使用的ratingMap
     * @param userId
     * @param ratingScore
     */
    private void maintainUserRating(Integer userId, Integer questionId, RatingScore ratingScore){
        String hashKey = RATING_OF_USER_HASH + "::" + userId;
        String zsetKey = TIME_OF_BEHAVIOR_ZSET + "::" + userId;
        String key = "key::" + questionId;
        //如果没有出错的话，hash和zset的size，包含的key都应该是一样的
        //这里没必要load from MySQL to Redis，因为实时推荐不需要考虑用户此前的做题记录
        checkUserRating(hashKey, zsetKey);
        Integer size = redisTemplate.hasKey(hashKey)?
                redisTemplate.boundHashOps(hashKey).size().intValue():0;
        if (redisTemplate.boundHashOps(hashKey).hasKey(key)){
            redisTemplate.boundHashOps(hashKey).put(key,
                    Math.max(Double.parseDouble(redisTemplate.boundHashOps(hashKey).get(key).toString())
                            , ratingScore.getValue()));
        }
        else {
            redisTemplate.boundHashOps(hashKey).put(key, ratingScore.getValue());
        }
        //如果没有出错的话，size大小不应该超过K
        if (size == K){
            //如果hash中有key，则只需要更新
            if (redisTemplate.boundHashOps(hashKey).hasKey(key)) {

            }
            //如果hash中不含有key，则需要剔除最旧的key，再将新key加入
            else {
                //pop操作已经包含了删除动作
                String oldKey = redisTemplate.boundZSetOps(zsetKey).popMin().getValue().toString();
                redisTemplate.boundHashOps(hashKey).delete(oldKey);
                redisTemplate.boundHashOps(hashKey).put(key, ratingScore.getValue());
            }
        }
        else {
            if (redisTemplate.boundHashOps(hashKey).hasKey(key)) {
                redisTemplate.boundHashOps(hashKey).put(key,
                        Math.max(Double.parseDouble(redisTemplate.boundHashOps(hashKey).get(key).toString())
                                , ratingScore.getValue()));
            }
            else {
                redisTemplate.boundHashOps(hashKey).put(key, ratingScore.getValue());
            }
        }
        redisTemplate.boundZSetOps(zsetKey).add(key, new Date().getTime());
    }

    private void checkUserRating(String hashKey, String zsetKey){
        Integer hashSize, zsetSize;
        if (!redisTemplate.hasKey(hashKey)) hashSize = 0;
        else hashSize = redisTemplate.boundHashOps(hashKey).size().intValue();
        if (!redisTemplate.hasKey(zsetKey)) zsetSize = 0;
        else zsetSize = redisTemplate.boundHashOps(hashKey).size().intValue();
        if (zsetSize != hashSize){
            System.out.println("Error! " + hashKey + "'s size not equals with "
            + zsetKey + "'s size!");
        }
        if ( zsetSize > K){
            System.out.println("Error! " + hashKey + "'s size > K!");
        }
    }

    /**
     * 计算两个问题之间的余弦相似度
     * @param qid1
     * @param qid2
     * @return
     */
    private Double calculateCosSimilarity(Integer qid1, Integer qid2){
        List<Double> emb1 = findQuestionEmb(qid1);
        List<Double> emb2 = findQuestionEmb(qid2);
        return calculateCosSimilarity(emb1, emb2);
    }

    private Double calculateCosSimilarity(List<Double> emb1, List<Double> emb2){
        Double dotxy = 0.0, _x = 0.0, _y = 0.0;
        for (int i = 0; i < embeddingSize; i++) {
            Double a = emb1.get(i);
            Double b = emb2.get(i);
            dotxy += a*b;
            _x += a*a;
            _y += b*b;
        }
        return dotxy/(Math.sqrt(_x)*Math.sqrt(_y));
    }

    private Double calculatePriority(Integer candidateQId, Map<Integer, Double> recentK){
        Double priority = 0.0;
        //sim_sum 表示 q 与 RK 中题目相似度大于最小阈值的个数；
        //incount 表示 RK 中与题目q 相似的、且本身“评分”较高（>=3）的题目个数；
        //recount 表示 RK 中与题目q 相似的、且本身“评分”较低（<3）的题目个数；
        Integer simSum = 0, inCount = 0, reCount = 0;
        for(Integer r:recentK.keySet()){
            Double similarity = calculateCosSimilarity(r, candidateQId);
            if (similarity < threshold)
                continue;
            simSum ++;
            if (recentK.get(r) >= 3) inCount++;
            else reCount++;
            priority += similarity*recentK.get(r);
        }
        System.out.println("priority:" + priority);
        System.out.println("inCount:" + inCount);
        System.out.println("reCount:" + reCount);
        if (simSum == 0) return 0.0;
//        else return priority/simSum + Math.log10(inCount) - Math.log10(reCount);
        else {
            priority = priority/simSum;
            if (inCount > 0) priority += Math.log10(inCount);
            if (reCount > 0) priority -= Math.log10(reCount);
            return priority;
        }
    }

    private Map<Integer, Double> hash2Map(String hashKey){
        Map<Integer, Double> map = new HashMap<>();
        if (!redisTemplate.hasKey(hashKey)) return map;
        Set<String> keys = redisTemplate.boundHashOps(hashKey).keys();
        for (String key:keys){
            String[] split = key.split("::");
            Integer qid = Integer.parseInt(split[1]);
            map.put(qid,Double.parseDouble(redisTemplate.
                    boundHashOps(hashKey).get(key).toString()));
        }
        return map;
    }

}
